package com.slipper.websocket;

import com.slipper.core.gson.GsonUtil;
import com.slipper.core.message.AbstractSubscriber;
import com.slipper.core.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class WebsocketSubscriber extends AbstractSubscriber {
    private static final Logger logger = LoggerFactory.getLogger(WebsocketSubscriber.class);
    private final BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
    private final WebSocketSession session;
    private Thread consumer;
    private boolean shutdown = false;

    public WebsocketSubscriber(WebSocketSession session) {
        this.session = session;
        startConsumer();
    }

    @Override
    public void update(Message message) {
        if (Message.CLOSE_MESSAGE == message) {
            this.shutdown();
            return;
        }

        try {
            queue.put(message);
        } catch (InterruptedException e) {
            logger.error("消息入栈被打断！");
        } catch (Exception e) {
            logger.error("消息发布出错！", e);
        }
    }

    private void shutdown() {
        this.shutdown = true;
        this.consumer.interrupt();
    }

    private void sendMessage(Message message) {
        try {
            synchronized(this.session) {
                if (!this.session.isOpen()) {
                    this.shutdown();
                    return;
                }
                this.session.sendMessage(new TextMessage(GsonUtil.toJson(message)));
            }
        } catch (Exception e) {
            logger.error("数据发送失败！", e);
        }
    }

    private void startConsumer() {
        this.consumer = new Thread(() -> {
            while (!this.shutdown) {
                try {
                    sendMessage(this.queue.take());
                } catch (InterruptedException e) {
                    if (!this.shutdown) {
                        logger.error("消息发送被打断！");
                    }
                } catch (Exception e) {
                    logger.error("消息发送出错！", e);
                }
            }
        });
        this.consumer.start();
    }
}
